package com.sui.bigdata.sml.web.service.impl;

import com.feidee.fdcommon.azkaban.AzkabanClient;
import com.feidee.fdcommon.azkaban.model.AbstractJobParam;
import com.feidee.fdcommon.azkaban.model.SparkJobParam;
import com.feidee.fdcommon.configuration.CustomConfiguration;
import com.feidee.fdcommon.util.ToolUtils;
import com.sui.bigdata.sml.web.misc.azkaban.AzkabanHelper;
import com.sui.bigdata.sml.web.repository.mapper.AzkabanMapper;
import com.sui.bigdata.sml.web.repository.model.AzFlow;
import com.sui.bigdata.sml.web.repository.model.AzJob;
import com.sui.bigdata.sml.web.repository.model.AzProject;
import com.sui.bigdata.sml.web.service.AzkabanService;
import com.sui.bigdata.sml.web.util.Constants;
import com.sui.bigdata.sml.web.util.SparkUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.DependsOn;
import org.springframework.stereotype.Service;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;

/**
 * @author songhaicheng
 * @date 2019/11/12 10:13
 * @description
 * @reviewer
 */
@Service
@DependsOn("disconfConfig")
@Slf4j
public class AzkabanServiceImpl implements AzkabanService {

    @Autowired
    private AzkabanMapper mapper;

    public static final String DELETED_TRUE = "01";

    @Override
    public String createProject(AzProject project) {
        AzkabanClient client = register(project.getUsername());

        List<SparkJobParam> sparkJobParams = project.getJobs();
        for (SparkJobParam sparkJobParam : sparkJobParams) {
            sparkJobParam.setJobClass(CustomConfiguration.getString(Constants.AZKABAN_SPARK_EXECUTOR_CLASS));
            sparkJobParam.setJobClassPath(CustomConfiguration.getString(Constants.AZKABAN_SPARK_EXECUTOR_CLASSPATH));
            sparkJobParam.setAlgLocalPath(CustomConfiguration.getString(Constants.AZKABAN_SPARKJOB_ALGLOCALPATH));
            sparkJobParam.setYarnUser(CustomConfiguration.getString(Constants.AZKABAN_YARNUSER));
            sparkJobParam.setSmlPath(CustomConfiguration.getString(Constants.AZKABAN_SPARKJOB_SMLPATH));
            sparkJobParam.setNotifyEmails(project.getEmails().trim());
            if (!ToolUtils.isNotNull(sparkJobParam.getAlgJar())) {
                sparkJobParam.setAlgJar(CustomConfiguration.getString(Constants.AZKABAN_SPARK_ALGPATH));
            }
            //queue 不再放开给调用方传，而采用配置或者根据用户名拼装的方式
            sparkJobParam.setQueue(SparkUtil.getQueue(project.getUsername()));
        }
        String projectId;
        List<AbstractJobParam> list = new ArrayList<>(sparkJobParams);
        try {
            projectId = client.addFlow(project.getName().trim(), project.getDescription(), list, project.getFailureAction(), project.getCron());
        } catch (Exception e) {
            log.error("failed to submit Azkaban project：", e);
            throw new RuntimeException("提交任务到 Azkaban 失败");
        }

        if (null != projectId) {
            log.info("get the projectId：{}", projectId);
            AzFlow flow = new AzFlow().setProjectId(Long.valueOf(projectId));
            if (ToolUtils.isNotNull(project.getName().trim())) {
                flow.setProjectName(project.getName().trim());
            }
            if (ToolUtils.isNotNull(project.getDescription())) {
                flow.setDescription(project.getDescription());
            }
            if (ToolUtils.isNotNull(project.getUsername())) {
                flow.setUsername(project.getUsername());
            }
            if (ToolUtils.isNotNull(project.getEmails())) {
                flow.setEmails(project.getEmails().trim());
            }
            if (ToolUtils.isNotNull(project.getFailureAction())) {
                flow.setFailureAction(project.getFailureAction());
            }
            if (ToolUtils.isNotNull(project.getCron())) {
                flow.setCron(project.getCron());
            }

            mapper.insertFlow(flow);
            String flowsId = flow.getId().toString();
            if (ToolUtils.isNotNull(flowsId)) {
                log.info("get the flowId：{}", flowsId);
                for (SparkJobParam sparkJobParam : sparkJobParams) {
                    AzJob job = new AzJob()
                            .setFlowId(Long.parseLong(flowsId))
                            .setJobId(sparkJobParam.getJobId())
                            .setParam(sparkJobParam.toString())
                            .setCreateTime(new Timestamp(System.currentTimeMillis()))
                            .setUpdateTime(new Timestamp(System.currentTimeMillis()));
                    mapper.insertJob(job);
                }
                return flowsId;
            }
        } else {
            log.error("the projectId is null");
        }
        return null;
    }

    @Override
    public void deleteProject(String name) {
        AzkabanClient client = register(AzkabanHelper.getAZUsername());

        name = name.trim();
        AzFlow flow = new AzFlow().setProjectName(name).setUsername(AzkabanHelper.getAZUsername());
        flow.setDeleted(DELETED_TRUE);

        try {
            client.deleteFlow(name);
            mapper.updateFlow(flow);
        } catch (Exception e) {
            log.error("delete Azkaban project failed：", e);
            throw new RuntimeException("从 Azkaban 删除项目失败：" + name);
        }
    }

    private AzkabanClient register(String userName) {
        log.info("user：{}，start to get Azkaban client..................", userName);
        try {
            return new AzkabanClient(userName, CustomConfiguration.getString("azkaban." + (userName != null ? userName : "") + ".password"));
        } catch (Exception e) {
            log.error("Azkaban initialize failed：", e);
            throw new RuntimeException("请检查 Azkaban 配置");
        }
    }
}
